package com.hong.springboot.listener;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSONObject;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.hong.springboot.api.IUserService;
import com.hong.springboot.entity.User;
import com.hong.springboot.utils.DataResponse;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

/**
 * @author: csh
 * @Date: 2021/3/16 11:14
 * @Description:用户监听
 */
@Service
@RocketMQMessageListener(consumerGroup ="${user_group_consumer}" , topic = "${user_topic}")
@Log4j2
public class UserListener implements RocketMQListener<MessageExt> {

    @Reference
    private IUserService userService;

    //存放缓存三天 分布式集群场景可以
    private static Cache<String,ConsumeConcurrentlyStatus> cache = CacheBuilder.newBuilder().expireAfterWrite(3, TimeUnit.DAYS).build();

    @Override
    public void onMessage(MessageExt msg) {
        if(null==msg || msg.getBody()==null || StringUtils.isEmpty(msg.getMsgId())){
            return;
        }
        ConsumeConcurrentlyStatus status = cache.getIfPresent(msg.getMsgId());
        //这两种消息才进行消费
        if(null==status || status.equals(ConsumeConcurrentlyStatus.RECONSUME_LATER)){
            String msgId = msg.getMsgId();
            log.info("msgId"+msgId);
            User user = RemotingSerializable.decode(msg.getBody(), User.class);
            log.info("springboot获取的用户信息{}", JSONObject.toJSONString(user));
            DataResponse<Boolean> save = userService.reallySave(user);
            log.info("添加结果{}",JSONObject.toJSONString(save));
            if(save==null || !save.getData()){
                log.info("添加失败，原因{}",JSONObject.toJSONString(save));
                cache.put(msgId, ConsumeConcurrentlyStatus.RECONSUME_LATER);
            }
            cache.put(msgId, ConsumeConcurrentlyStatus.CONSUME_SUCCESS);
        }else{
            log.info("该消息{}已消费过{}",JSONObject.toJSONString(msg),JSONObject.toJSONString(status));
        }

    }

}
